package net.bwie.realtime.jtp.dwd.log.job;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.jtp.common.utils.DateTimeUtil;
import net.bwie.realtime.jtp.common.utils.KafkaUtil;
import net.bwie.realtime.jtp.dwd.log.clickhouse.JtpLogClickHouse;
import net.bwie.realtime.jtp.dwd.log.function.IsNewprocessFuncation;
import net.bwie.realtime.jtp.dwd.log.function.LogSplitProcessFuncation;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class LogEtlJob {
    public static void main(String[] args) throws Exception {
        //1. 执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度
        env.setParallelism(1);

        //2. 获取数据源
        DataStream<String> kafkaDataStream = KafkaUtil.consumerKafka(env, "topic-log");

        //3. 数据转换
        processlog(kafkaDataStream);
        //4. 数据输出




        //触发执行
        env.execute();

    }

    /**
     * 对实时获取APP流量日志数据进行ETL处理，并且进行分流，存储Kafka消息队列
     *      1-数据清洗
     *      2-新老访客状态标记修复
     *      3-数据分流
     *      4-存储数据
     */
    private static void processlog(DataStream<String> kafkaDataStream) {
        //1. 数据清洗
        DataStream<String> cleandStream = logcleand(kafkaDataStream);
        //2. 新老访客状态标记修复
        DataStream<String> isNewStream = processIsNew(cleandStream);
        //3. 数据分流
        SingleOutputStreamOperator<String> splitStream = splitStream(isNewStream);
        //4. 存储数据
        //KafkaUtil.producerKafka(splitStream, "dwd_log_page_view_log");

        //todo 保存到clickhouse
        sinkToClickhouse(splitStream);

    }

    private static void sinkToClickhouse(SingleOutputStreamOperator<String> splitStream) {

        JtpLogClickHouse.savePageLog(splitStream);

        JtpLogClickHouse.saveStartLog(splitStream.getSideOutput(new OutputTag<String>("start-log"){}));

    }


    // 分流
    private static SingleOutputStreamOperator<String> splitStream(DataStream<String> isNewStream) {
        OutputTag<String> errorLog = new OutputTag<String>("error-log"){};
        OutputTag<String> startLog = new OutputTag<String>("start-log"){};
        OutputTag<String> displayLog = new OutputTag<String>("display-log"){};
        OutputTag<String> actionLog = new OutputTag<String>("action-log"){};

        SingleOutputStreamOperator<String> process = isNewStream.process(
               new LogSplitProcessFuncation(errorLog, startLog, displayLog, actionLog)
                );

        DataStream<String> errorStream = process.getSideOutput(errorLog);
        KafkaUtil.producerKafka(errorStream, "dwd_traffic-error-log");
        DataStream<String> startStream = process.getSideOutput(startLog);
        KafkaUtil.producerKafka(startStream, "dwd_traffic-start-log");
        DataStream<String> displayStream = process.getSideOutput(displayLog);
        KafkaUtil.producerKafka(displayStream, "dwd_traffic-display-log");
        DataStream<String> actionStream = process.getSideOutput(actionLog);
        KafkaUtil.producerKafka(actionStream, "dwd_traffic-action-log");



        return process;

    }



    // 数据清洗
    private static DataStream<String> logcleand(DataStream<String> kafkaDataStream) {

        OutputTag<String> dirtyLog = new OutputTag<String>("dirty-log"){};
        SingleOutputStreamOperator<String> cleandStream = kafkaDataStream.process(new ProcessFunction<String, String>() {
            @Override
            public void processElement(String value, ProcessFunction<String, String>.Context ctx, Collector<String> out) throws Exception {

                try {
                    // 解析JSON
                    JSON.parse(value);
                    out.collect(value);
                } catch (Exception e) {
                    ctx.output(dirtyLog, value);
                }
            }
        });
        KafkaUtil.producerKafka(cleandStream.getSideOutput(dirtyLog),"dwd-traffic-dirty-log");
        return cleandStream;
    }



    // 新老访客状态标记修复
    private static DataStream<String> processIsNew(DataStream<String> cleandStream) {

        KeyedStream<String, String> midStream = cleandStream.keyBy(new KeySelector<String, String>() {
            @Override
            public String getKey(String value) throws Exception {
                return JSON.parseObject(value).getJSONObject("common").getString("mid");
            }
        });

        SingleOutputStreamOperator<String> isNewStream = midStream.process(new IsNewprocessFuncation());


        return isNewStream;
    }




}
